package com.battcn.elasticsearch.configuration;


import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.elasticsearch.client.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;

/**
 * es配置
 *
 * @author videomonster
 */
@Slf4j
@Configuration
public class ElasticsearchConfig {

    @Bean
    public RestClientBuilder restClientBuilder(EsProperties esProperties, HttpProperties httpProperties) {
        // 获取es节点的host信息
        String[] clusterNodes = esProperties
                .getClusterNodes()
                .replace(" ", "")
                .split(",");
        HttpHost[] httpHosts = new HttpHost[clusterNodes.length];
        for (int i = 0; i < clusterNodes.length; i++) {
            String[] node = clusterNodes[i].split(":");
            httpHosts[i] = new HttpHost(node[0], Integer.parseInt(node[1]));
        }
        RestClientBuilder.HttpClientConfigCallback httpClientConfigCallback = httpClientBuilder ->
                httpClientBuilder.setDefaultIOReactorConfig(
                        IOReactorConfig.custom()
                                .setIoThreadCount(Runtime.getRuntime().availableProcessors() * 5)
                                .build())
                        .setMaxConnTotal(httpProperties.getMaxTotal())
                        .setMaxConnPerRoute(httpProperties.getDefaultMaxPerRoute());
        return RestClient.builder(httpHosts)
                //.setMaxRetryTimeoutMillis(httpProperties.getReadTimeout())
                .setHttpClientConfigCallback(httpClientConfigCallback)
                .setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder
                        .setConnectionRequestTimeout(httpProperties.getConnectionRequestTimeout())
                        .setConnectTimeout(httpProperties.getConnectTimeout())
                        .setSocketTimeout(httpProperties.getReadTimeout())
                ).
                        setFailureListener(new RestClient.FailureListener() {

                            @Override
                            public void onFailure(Node node) {
                                log.error("elasticsearch server occur error.");
                                super.onFailure(node);
                            }
                        });
    }

    @Bean
    public RestHighLevelClient restHighLevelClient(RestClientBuilder restClientBuilder) throws IOException {
        RestHighLevelClient client = new RestHighLevelClient(restClientBuilder);
        boolean ping = client.ping(RequestOptions.DEFAULT);
        if (ping) {
            log.info("elasticesearch server connected");
        }
        return client;
    }
}
